package flink.http_connect;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.util.EntityUtils;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.function.Supplier;

public class AsyncHttpFunction extends RichAsyncFunction<String, R> {

    CloseableHttpAsyncClient httpclient = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        RequestConfig requestConfig = RequestConfig.custom()
                .setSocketTimeout(3000)
                .setConnectTimeout(3000) //设置HttpClient连接池超时时间
                .build();

        httpclient = HttpAsyncClients.custom()
                .setMaxConnTotal(20) //连接池最大连接数
                .setDefaultRequestConfig(requestConfig)
                .build();

        httpclient.start();
    }


    // 异步处理函数的主执行方法
    @Override
    public void asyncInvoke(String line, ResultFuture<R> resultFuture) throws Exception {
        try {
            //1、创建httpGet请求
            HttpGet httpGet = new HttpGet("http://127.0.0.1:25001/parkLive/flush?param="+line);//http接口
            // 2、提交httpclient异步请求，获取异步请求的future对象
            Future<HttpResponse> future = httpclient.execute(httpGet, null);
            //3、解析HttpResponse
            HttpResponse response = future.get();
            R r;
            if (response.getStatusLine().getStatusCode() == 200) {
                HttpEntity entity = response.getEntity();
                String result = EntityUtils.toString(entity);
                r = JSON.parseObject(result,R.class);
            }else{
                r = R.fail();
            }
            //4、返回
            resultFuture.complete(Collections.singleton(r));
        } catch (Exception e) {
            e.printStackTrace();
            resultFuture.complete(Collections.singleton(null));
        }
    }

    // 关闭连接
    @Override
    public void close() throws Exception {
        httpclient.close();
    }
}

